package cn.xuexiyuan.flinkstudy.transformation;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description:  各种分区
 *
 * @Author 左龙龙
 * @Date 21-3-24
 * @Version 1.0
 **/
public class TransformationDemo05 {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> ds1 = env.readTextFile("data/input/word1.txt");

        // 2.transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = ds1.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            for (String token : s.split("\\W+")) {
                collector.collect(Tuple2.of(token, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));

        DataStream<Tuple2<String, Integer>> result1 = tupleDS.global();
        DataStream<Tuple2<String, Integer>> result2 = tupleDS.broadcast();
        DataStream<Tuple2<String, Integer>> result3 = tupleDS.forward();
        DataStream<Tuple2<String, Integer>> result4 = tupleDS.shuffle();
        DataStream<Tuple2<String, Integer>> result5 = tupleDS.rebalance();
        DataStream<Tuple2<String, Integer>> result6 = tupleDS.rescale();
        DataStream result7 = tupleDS.partitionCustom(new MyPartitioner(), t -> t.f0);


        // 3.sink
        result1.print("global");
        result2.print("broadcast");
        result3.print("forward");
        result4.print("shuffle");
        result5.print("rebalance");
        result6.print("rescale");
        result7.print("partitionCustom");

        // 4.excute
        env.execute("TransformationDemo05");
    }
}


class MyPartitioner implements Partitioner{
    @Override
    public int partition(Object o, int i) {
        return 0;
    }
}